#pragma once
#include <string>
#include <vector>
#include <iostream>
#include "../mqcommom/helper.hpp"
#include "../mqcommom/msg.pb.h"
#include <memory>
#include <functional>
#include <mutex>
#include <unordered_map>
/**************************
 *消费者模块
 *    队列管理:消费者的管理是以队列进行管理的，因为在某一个队列上有数据之后就需要推送到指定的消费者
 *   回调函数，用于上层提供处理业务
 *   自动删除标志，消息发送后是否保留
 *
 *以队列管理:
 *  vector<consumer>用于存储选择的下标
 *  rr_seq 自增，顺序选择
 *  qname 队列名称
 */

namespace ns_rabbitmq
{
    using ConsumCallback = std::function<void(const std::string, BasicProperties *, const std::string)>;
    class Consumer
    {
    public:
        Consumer(const std::string &tag, const std::string &qname, bool auto_delete, ConsumCallback cb)
            : _tag(tag), _qname(qname), _auto_delete(auto_delete), _cb(cb)
        {
        }
        ~Consumer() {}

        std::string _tag;   // 标签,消费者的标识
        std::string _qname; // 订阅哪一个队列
        bool _auto_delete;
        ConsumCallback _cb;
    };

    using ConsumerPtr = std::shared_ptr<Consumer>;

    class QueueConsumer
    {
    public:
        QueueConsumer(const std::string &qname)
            : _qname(qname), _rr_seq(0)
        {
        }
        QueueConsumer() {}

        ConsumerPtr Create(const std::string &tag, const std::string &qname, bool auto_delete,const  ConsumCallback &cb)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            for (auto it = _consumers.begin(); it != _consumers.end(); ++it)
            {
                if ((*it)->_tag == tag)
                {
                    ILOG("向指定队列%s,创建消费者%s失败,因为已经存在!", qname.c_str(), tag.c_str());
                    return ConsumerPtr();
                }
            }
            ConsumerPtr consumptr = std::make_shared<Consumer>(tag, qname, auto_delete, cb);
            _consumers.push_back(consumptr);
            DLOG("向consumers中添加消费者%s",tag.c_str());
            return consumptr;
        }

        void Remove(const std::string &tag)
        {
            // 遍历所有的消费者删除
            std::unique_lock<std::mutex> lock(_mutex);
            for (auto it = _consumers.begin(); it != _consumers.end(); ++it)
            {
                if ((*it)->_tag == tag)
                {
                    _consumers.erase(it);
                    return;
                }
            }
        }

        // 采用rr轮转，负载均衡的选择消费者
        ConsumerPtr Choose()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            if (!_consumers.size())
                return ConsumerPtr();
            int index = _rr_seq % _consumers.size();
            ++_rr_seq;
            return _consumers[index];
        }
        bool IsEmpty()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _consumers.size() == 0;
        }

        bool IsExistsConsumer(const std::string &tag)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            for (auto it = _consumers.begin(); it != _consumers.end(); ++it)
            {
                if ((*it)->_tag == tag)
                {
                    return true;
                }
            }
            return false;
        }

        void Clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _consumers.clear();
            _rr_seq = 0;
        }

    private:
        std::mutex _mutex;
        std::string _qname;
        uint64_t _rr_seq;
        std::vector<ConsumerPtr> _consumers;
    };

    using QueueConsumerPtr = std::shared_ptr<QueueConsumer>;
    class ConsumerMananger
    {
    public:
        ConsumerMananger() {}
        void InitQueueConsumer(const std::string &qname)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _consumerqueues.find(qname);
            if (it != _consumerqueues.end())
            {
                return;
            }
            QueueConsumerPtr qconsumptr = std::make_shared<QueueConsumer>(qname);
            _consumerqueues.insert(std::make_pair(qname, qconsumptr));
        }

        void DestroyQueueConsumer(const std::string &qname)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _consumerqueues.erase(qname);
        }

        ConsumerPtr Create(const std::string &tag, const std::string &qname, bool auto_delete, ConsumCallback cb)
        {
            QueueConsumerPtr qconsumptr;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _consumerqueues.find(qname);
                if (it == _consumerqueues.end())
                {
                    DLOG("找不到队列%s的管理句柄", qname.c_str());
                    return ConsumerPtr();
                }
                qconsumptr = it->second;
            }
            return qconsumptr->Create(tag, qname, auto_delete, cb);
        }

        void Remove(const std::string &tag, const std::string &qname)
        {
            QueueConsumerPtr qconsumptr;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _consumerqueues.find(qname);
                if (it == _consumerqueues.end())
                {
                    DLOG("找不到队列%s的管理句柄", qname.c_str());
                    return;
                }
                qconsumptr = it->second;
            }
            qconsumptr->Remove(tag);
        }

        ConsumerPtr Chose(const std::string &qname)
        {
            QueueConsumerPtr qconsumptr;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _consumerqueues.find(qname);
                if (it == _consumerqueues.end())
                {
                    DLOG( "找不到队列%s的管理句柄", qname.c_str());
                    return ConsumerPtr();
                }
                qconsumptr = it->second;
            }
            return qconsumptr->Choose();
        }

        bool IsQueueEmpty(const std::string &qname)
        {
            QueueConsumerPtr qconsumptr;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _consumerqueues.find(qname);
                if (it == _consumerqueues.end())
                {
                    DLOG("找不到队列%s的管理句柄", qname.c_str());
                    return false;
                }
                qconsumptr = it->second;
            }
            return qconsumptr->IsEmpty();
        }

        bool IsConsumerExists(const std::string &tag, const std::string &qname)
        {
            QueueConsumerPtr qconsumptr;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _consumerqueues.find(qname);
                if (it == _consumerqueues.end())
                {
                    DLOG("找不到队列%s的管理句柄", qname.c_str());
                    return false;
                }
                qconsumptr = it->second;
            }
            return qconsumptr->IsExistsConsumer(tag);

        }

        void Clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _consumerqueues.clear();
        }

    private:
        std::mutex _mutex;
        std::unordered_map<std::string, QueueConsumerPtr> _consumerqueues;
    };
    using ConsumerManangerPtr=std::shared_ptr<ConsumerMananger>;
}